/**
 * @author: yyq
 * @date: 2025年3月13日 10:10:10
 * @description: 
 */
package com.sheepone.monitor.server.handler.connect;

import java.util.List;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.sheepone.monitor.anno.ChannelHandlerOrder;
import com.sheepone.monitor.enums.SerializationAlgorithmsEnum;
import com.sheepone.monitor.enums.SheeponePackageTypeEnum;
import com.sheepone.monitor.handler.AbstractSheeponeContentInboundHandler;
import com.sheepone.monitor.proto.HeartBeatCheckMessage;
import com.sheepone.monitor.proto.HeartBeatReportMessage;
import com.sheepone.util.time.TimeFormatter;
import com.sheepone.webdev.log.WebDevLogger;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import io.netty.channel.ChannelHandler.Sharable;

/**
 * @author: yyq
 * @date: 2025年3月13日 10:10:10
 * @description:
 *               1、处理来自客户端的HeartBeatAckMessage
 */
@Sharable
@ChannelHandlerOrder(Integer.MIN_VALUE + 1)
public class ServerHeartbeatHandler extends AbstractSheeponeContentInboundHandler<HeartBeatReportMessage> {

    public ServerHeartbeatHandler() {
        super(SheeponePackageTypeEnum.HEART_BEAT_REPORT, SerializationAlgorithmsEnum.PROTOBUF);
    }

    private static ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(
            1, new CustomThreadFactory("heartbeat-pool"));

    private static WebDevLogger logger = WebDevLogger.getLogger(ServerHeartbeatHandler.class);

    static {
        scheduler.scheduleAtFixedRate(() -> {
            List<Channel> channels = ConnectionManager.getUnhealthyChannels(2, TimeUnit.MINUTES);
            channels.forEach(channel -> {
                if (channel.isActive()) {
                    channel.writeAndFlush(
                            HeartBeatCheckMessage.newBuilder().setTime(System.currentTimeMillis()).build());
                } else {
                    ConnectionManager.remove(channel.id());
                }
            });
        }, 0, 5, TimeUnit.MINUTES);
    }

    @Override
    public HeartBeatReportMessage fromBytes(byte[] bytes, SerializationAlgorithmsEnum serialization) throws Exception {
        return HeartBeatReportMessage.parseFrom(bytes);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, HeartBeatReportMessage msg) throws Exception {
        ChannelId id = ctx.channel().id();
        ConnectionMeta meta = ConnectionManager.get(id);
        long time = msg.getTime();
        meta.setLastHeartbeatTime(time);
        logger.info("receive heartbeat for {}", TimeFormatter.formatTimeStamp(time));
    }
}

class CustomThreadFactory implements ThreadFactory {
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    CustomThreadFactory(String poolName) {
        namePrefix = poolName + "-thread-";
    }

    public Thread newThread(Runnable r) {
        return new Thread(r, namePrefix + threadNumber.getAndIncrement());
    }
}
